#include <stdlib.h>
#include <string.h>

#include "hidb.h"
#include "md5.h"

#include <time.h>
#include <sys/time.h>
#include <assert.h>

struct timeval start;
struct timeval end;
double cost;
#define CASE_START  gettimeofday(&start, NULL)
#define CASE_END    gettimeofday(&end, NULL)
#define CASE_COST   do{ cost = (double)(end.tv_sec - start.tv_sec) \
						+(double)(end.tv_usec - start.tv_usec)/1000000;\
						printf("cst: %f, spd : %f\n", cost, (double)i/cost);\
					}while(0)

/*
 *	[meta{table}]
 *	Url | Status | Signature | ModifedTime | FetchTime | FetchInterval | Score
 *	[data{hash}]
 *	Signature | Text
*/
static const char * hifiled[] = {
	/* URL is key */
	"stat",		/* F_STAT */
	"sign",		/* F_SIGN */
	"mtime",	/* F_MTIME */
	"ftime",	/* F_FTIME */
	"fintv",	/* F_FINTV */
	"score",	/* F_SCORE */
};

static unsigned int dohash(const char *kbuf, int kbytes)
{
	unsigned int hash = 751;
	const char *rp = kbuf + kbytes;
	while (kbytes --) {
		hash = (hash * 31) ^ *(char *)--rp;
	}
	return hash;
}

static char * domd5(const char *text, int textlen, char *sign)
{
	assert(sign);
	if (!text || textlen <= 0) {
		sign[0] = '\0';
		return sign;
	}

	int i;
	md5_state_t ms;
	md5_init(&ms);
	md5_append(&ms, (md5_byte_t *)text, textlen);
	unsigned char digest[16];
	md5_finish(&ms, (md5_byte_t *)digest);
	char *wp = sign;
	for(i = 0; i < 16; i++){
		wp += sprintf(wp, "%02x", digest[i]);
	}
	*wp = '\0';
	return sign;
}


int hidb_vgroup_free(VGROUP * p)
{
	if (!p) {
		return 1;
	}

	if (!p->vnum) {
		return 2;
	}
	int i;
	for (i = 0; i < p->vnum; i ++) {
		VNODE * v = p->vnodes + i;
		if(v->dbp){
			tcrdbdel(v->dbp);
		}
	}
	memset(p, 0, sizeof(struct VGROUP));
	return 0;
}

int hidb_free(HIDB * p)
{
	if (!p) {
		return 1;
	}
	hidb_vgroup_free(&p->metadb);
	hidb_vgroup_free(&p->datadb);
	free(p);
	return 0;
}

HIDB * hidb_new(int num, VNODE * nodes)
{
	if (num <= 0 || num > HIDB_VNODE_MAX*3 || !nodes) {
		return NULL;
	}
	HIDB * p = (HIDB *)malloc(sizeof(struct HIDB));
	if (p == NULL) {
		return NULL;
	}

	while (num --) {
		VNODE * n = nodes ++;
		VGROUP * db = NULL;
		switch (n->type) {
		case HIDB_META: db = &p->metadb; break;
		case HIDB_DATA: db = &p->datadb; break;
		default:
			hidb_free(p);
			return NULL;
		}

		if (db->vnum == HIDB_VNODE_MAX) {
			hidb_free(p);
			return NULL;
		}
		VNODE * v = db->vnodes + (db->vnum ++);
		memcpy(v, n, sizeof(struct VNODE));

		v->dbp = tcrdbnew();
		if(v->dbp == NULL || !tcrdbopen(v->dbp, v->host, v->port) ||
				tcrdbexpr(v->dbp) == NULL){
			hidb_free(p);
			return NULL;
		}
	}

	if (!p->metadb.vnum || !p->datadb.vnum) {
		hidb_free(p);
		return NULL;
	}
	return p;
}

int fval2map(TCMAP * map, FVAL * fval)
{
	assert(map && fval);
	char s[HIDB_VAL_SIZE];
	//stat
	snprintf(s, HIDB_VAL_SIZE - 1, "%d", fval->stat);
	tcmapput2(map, hifiled[F_STAT], s);
	//sign
	tcmapput2(map, hifiled[F_SIGN], fval->sign);
	//mtime
	snprintf(s, HIDB_VAL_SIZE - 1, "%ld", fval->mtime);
	tcmapput2(map, hifiled[F_MTIME], s);
	//ftime
	snprintf(s, HIDB_VAL_SIZE - 1, "%ld", fval->ftime);
	tcmapput2(map, hifiled[F_FTIME], s);
	//fintv
	snprintf(s, HIDB_VAL_SIZE - 1, "%d", fval->fintv);
	tcmapput2(map, hifiled[F_FINTV], s);
	//score
	snprintf(s, HIDB_VAL_SIZE - 1, "%d", fval->score);
	tcmapput2(map, hifiled[F_SCORE], s);

	return 0;
}

int map2fval(FVAL * fval, TCMAP * map)
{
	assert(map && fval);
	const char * s; 
	//stat
	if (!(s = tcmapget2(map, hifiled[F_STAT]))) {
		return -1;
	}
	fval->stat = atoi(s);
	//sign
	if (!(s = tcmapget2(map, hifiled[F_SIGN]))) {
		return -2;
	}
	strncpy(fval->sign, s, HIDB_SIGN_SIZE - 1);
	//mtime
	if (!(s = tcmapget2(map, hifiled[F_MTIME]))) {
		return -3;
	}
	fval->mtime = atol(s);
	//ftime
	if (!(s = tcmapget2(map, hifiled[F_FTIME]))) {
		return -4;
	}
	fval->ftime = atol(s);
	//fintv
	if (!(s = tcmapget2(map, hifiled[F_FINTV]))) {
		return -5;
	}
	fval->fintv = atol(s);
	//score
	if (!(s = tcmapget2(map, hifiled[F_SCORE]))) {
		return -6;
	}
	fval->score = atol(s);
	return 0;
}

char * hidb_getbysign(const HIDB * p, const char * sign, int * datalenp)
{
	//data vnode depend on sign
	const VNODE * vdata = HASHVND(sign, strlen(sign), p->datadb);
	assert(vdata->type == HIDB_DATA && vdata->dbp);

	char * data = tcrdbget(vdata->dbp, sign, strlen(sign), datalenp);

	return data;
}

char * hidb_getbyurl(const HIDB * p, const char * url, int * datalenp)
{
	int urllen = strlen(url);

	//meta vnode depend on url
	const VNODE * vmeta = HASHVND(url, urllen, p->metadb);
	assert(vmeta->type == HIDB_META && vmeta->dbp);

	FVAL fval = FVAL_INIT;
	TCMAP * map = tcrdbtblget(vmeta->dbp, url, urllen);
	if (!map) {
		return NULL;
	} else if (map2fval(&fval, map) < 0) {
		tcmapdel(map);
		return NULL;
	}
	tcmapdel(map);

	if (fval.stat == HIDB_NULL) {
		return NULL;
	}

	return hidb_getbysign(p, fval.sign, datalenp);
}

/* 
 *	text == NULL && url not exist
 *		stat = FREASH
 *		put meta
 *	text == NULL && url exist
 *		stat = ERR
 *		put meta
 *	text != NULL && url exist stat = OK
 *		stat = OK
 *		put meta
 *		put data if sign changed
*/
int hidb_put(HIDB * p, const char * url, const char * text, int textlen)
{
	if (!p || !url) {
		return -1;
	}

	int urllen = strlen(url);
	if (urllen < 0 || !p->metadb.vnum || !p->datadb.vnum) {
		return -2;
	}

	FVAL oldval = FVAL_INIT;
	FVAL newval = FVAL_INIT;

	/* metadb: insert or update meta */
	//meta vnode depend on url
	const VNODE * vmeta = HASHVND(url, urllen, p->metadb);
	assert(vmeta->type == HIDB_META && vmeta->dbp);

	/* metadb: get old page meta */
	TCMAP * map = tcrdbtblget(vmeta->dbp, url, urllen);
	if ((!map || map2fval(&oldval, map) < 0) && text <= 0) {
		newval.stat = HIDB_FRESH;
		newval.sign[0] = 0;
		newval.mtime = 0;
		newval.ftime = time(NULL) + HIDB_DEF_INTV;
		newval.fintv = HIDB_DEF_INTV;
		newval.score = HIDB_DEF_SCORE;
	} else if (textlen >= 0) {
		newval.stat = HIDB_OK;
		domd5(text, textlen, newval.sign);
		if (!strncmp(newval.sign, oldval.sign, HIDB_SIGN_SIZE)) {
			newval.mtime = oldval.mtime;
		} else {
			newval.mtime = time(NULL);
		}
		newval.ftime = time(NULL) + HIDB_DEF_INTV;
		newval.fintv = HIDB_DEF_INTV;
		newval.score = HIDB_DEF_SCORE;
	} else {
		newval.stat = HIDB_ERR;
		newval.sign[0] = 0;
		newval.mtime = oldval.mtime;
		newval.ftime = time(NULL) + HIDB_DEF_INTV;
		newval.fintv = HIDB_DEF_INTV;
		newval.score = HIDB_DEF_SCORE;
	}
	if (map)	tcmapdel(map);

	TCMAP *cols = tcmapnew();
	fval2map(cols, &newval);
	if (!tcrdbtblput(vmeta->dbp, url, urllen, cols)) {
		tcmapdel(cols);
		return -3;
	}
	tcmapdel(cols);

	/* datadb: insert while new sign isn't exsit in data */
	//fresh link or fetch nothing
	if (newval.stat == HIDB_FRESH || newval.stat == HIDB_ERR) {
		return 1;
	}
	//nothing modify
	if (newval.stat == HIDB_OK && !strncmp(oldval.sign, newval.sign, HIDB_SIGN_SIZE-1)) {
		return 2;
	}
	//data vnode depend on new sign
	const VNODE * vdata = HASHVND(newval.sign, strlen(newval.sign), p->datadb);
	assert(vdata->type == HIDB_DATA && vdata->dbp);

	if (!tcrdbput(vdata->dbp, newval.sign, strlen(newval.sign), text, textlen)) {
		return -4;
	}

	/* datadb: delete while old sign isn't exsit in meta */
	if (!oldval.sign[0]) {
		return 3;
	}

	RDBQRY *qry = tcrdbqrynew(vmeta->dbp);
	tcrdbqryaddcond(qry, hifiled[F_SIGN], RDBQCSTREQ, oldval.sign);
	tcrdbqrysetlimit(qry, 1, 0);
	TCLIST * lst = tcrdbqrysearch(qry);
	if (tclistnum(lst) > 0) {
		free(lst);
		tcrdbqrydel(qry);
		return 4;
	}

	free(lst);
	tcrdbqrydel(qry);

	//data vnode depend on old sign
	vdata = HASHVND(oldval.sign, strlen(oldval.sign), p->datadb);
	assert(vdata->type == HIDB_DATA && vdata->dbp);

	if (!tcrdbout2(vdata->dbp, oldval.sign)) {
		return 5;
	}

	return 0; 
}

TCLIST * hidb_qry(HIDB * p, enum HI_FILED f, const char * start, const char * end, int limit)
{
	if (!p || !(start || end)) {
		return NULL;
	}

	TCLIST * merge = tclistnew();

	int i;
	for(i = 0; i < p->metadb.vnum; i++){
		VNODE * v = p->metadb.vnodes + i;

		RDBQRY *qry = tcrdbqrynew(v->dbp);
		if (start)	tcrdbqryaddcond(qry, hifiled[f], RDBQCNUMGE, start);
		if (end)	tcrdbqryaddcond(qry, hifiled[f], RDBQCNUMLE, end);
		if (limit > 0)	tcrdbqrysetlimit(qry, 1, 0);

		TCLIST * res = tcrdbqrysearch(qry);
		for(i = 0; i < tclistnum(res); i++){
			int rsiz;
			const char * rbuf = tclistval(res, i, &rsiz);
			tclistpush(merge, rbuf, rsiz);
		}
		tclistdel(res);
		tcrdbqrydel(qry);
	}
	return merge;
}

#ifdef HITBL_MAIN
int main(int argc, char ** argv)
{
	int total = 1;
	int size = 1;
	if (argc > 1 && argv[1]) {
		total = atoi(argv[1]);
	}
	if (argc > 2 && argv[2]) {
		size = atoi(argv[2]);   //kByte
	}

	const char * cfg_path = "hi.ini";
	CFGMAP * cfgp = cfg_new();
	if (cfgp == NULL) {
		return 1;
	}
	if (cfg_load(cfgp, cfg_path) < 0) {
		return 2;
	}
	cfg_dump(cfgp);
	int meta_vnum = cfg_get_int(cfgp, "hidb:meta_vnum", 0);
	if (meta_vnum < 1 || meta_vnum >= HIDB_VNODE_MAX) {
		return 3;
	}
	int data_vnum = cfg_get_int(cfgp, "hidb:data_vnum", 0);
	if (data_vnum < 1 || data_vnum >= HIDB_VNODE_MAX) {
		return 4;
	}

	int i, j;
	VNODE vnodes[HIDB_VNODE_MAX*2];
	const char * group = "meta";
	int type = HIDB_META;
	for (i = 0, j = 0; i<meta_vnum + data_vnum; i ++, j ++) {
		VNODE * v = vnodes + i;
		if (i == meta_vnum) {
			group = "data";
			type = HIDB_DATA;
			j = 0;
		}
		char param[32]; 
		sprintf(param, "%s%d", group, j);
		if(!cfg_isset_section(cfgp, param)){
			return 4;
		}
		sprintf(param, "%s%d:host", group, j);
		const char * host = cfg_get_str(cfgp, param, NULL);
		if (!host) {
			return 5;
		}
		strncpy(v->host, host, HIDB_HOST_SIZE-1);
		sprintf(param, "%s%d:port", group, j);
		v->port = cfg_get_int(cfgp, param, 0);
		v->type = type;
	}

	HIDB * p = hidb_new(meta_vnum + data_vnum, vnodes);
	if (p == NULL) {
		return 6;
	}

	char * pkbuf = (char *)malloc(32);
	char * pvbuf = (char *)malloc(size);
	fprintf(stderr, "--- write %d %dk---\n", total, size);
	CASE_START;
	for(i=1; i<=total; i++){
		sprintf(pkbuf, "http://kaixin%09d.com/", i);
		sprintf(pvbuf, "%09d", i);
		if(hidb_put(p, pkbuf, pvbuf, size<strlen(pvbuf)?strlen(pvbuf):size) < 0){
			return 7;
		}
		if(i%10000 == 0)
			fprintf(stderr, ". ");
		if(i%100000 == 0){
			fprintf(stderr, "\n");
			CASE_END;
			CASE_COST;
		}
	}
	fprintf(stderr, "--- query ---\n");
	/*	CASE_START;
		TCLIST * res = hidb_qry(p, "1274423197", "1274423198");
		for(i = 0; i < tclistnum(res); i++){
		int rsiz;
		char * rbuf = (char *)tclistval(res, i, &rsiz);
		fprintf(stderr, "%s\n", rbuf);
		}
		CASE_END;
		CASE_COST;
		tclistdel(res);
	 */
	int len;
	char *buf = hidb_getbyurl(p, "http://kaixin000000001.com/", &len);
	printf("get by url %s\n", buf);
	hidb_free(p);
	return 0;
}
#endif

